"""
邮件同步任务管理器
使用消息队列方式管理所有邮箱账户的后台同步任务
"""
import asyncio
from typing import Dict, Optional
from api.utils.helpers import get_beijing_time
from api.utils.logger import get_logger
from api.database import SessionLocal
from api.model.email_account import EmailAccount
from api.service.email_service import EmailService
from api.utils.cache import cache

logger = get_logger("sync_manager")
sync_logger = get_logger("sync")


class EmailSyncManager:
    """邮件同步任务管理器（单例模式）"""
    
    _instance = None
    _lock = asyncio.Lock()
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self):
        # 避免重复初始化
        if hasattr(self, '_initialized'):
            return
        
        self._initialized = True
        self.managed_accounts: set[int] = set()
        self.main_loop_task: Optional[asyncio.Task] = None
        self.sync_locks: Dict[int, asyncio.Lock] = {}  # account_id -> Lock (防止同一账户并发同步)
        self.sync_semaphore = asyncio.Semaphore(5) # 并发同步任务数限制
        self.running = False
        self.sync_interval = 300  # 默认5分钟同步一次
    
    async def start(self):
        """启动同步管理器"""
        if self.running:
            return
        
        self.running = True
        
        # ========== 启动时状态恢复逻辑 ==========
        # 使用重试机制批量重置异常状态
        max_retries = 10
        base_delay = 1.0
        
        for attempt in range(max_retries):
            db = None
            try:
                db = SessionLocal()
                
                # 查找所有处于同步中状态的账户
                stuck_accounts = db.query(EmailAccount).filter(
                    EmailAccount.sync_status == "syncing"
                ).all()
                
                if stuck_accounts:
                    sync_logger.info(f"发现 {len(stuck_accounts)} 个账户状态异常（状态：syncing），正在重置...")
                    for account in stuck_accounts:
                        sync_logger.info(
                            f"重置账户 {account.email} (ID: {account.id}) 的同步状态: "
                            f"syncing -> idle (最后同步: {account.last_sync})"
                        )
                        account.sync_status = "idle"
                    db.commit()
                    sync_logger.info(f"已重置 {len(stuck_accounts)} 个账户的同步状态")
                
                # 加载所有账户
                accounts = db.query(EmailAccount).all()
                db.close()
                db = None
                
                for account in accounts:
                    self.managed_accounts.add(account.id)
                
                # 启动中心化同步循环
                self.main_loop_task = asyncio.create_task(self._central_sync_loop())
                
                sync_logger.info(f"同步管理器已启动，管理 {len(self.managed_accounts)} 个账户")
                if accounts:
                    sync_logger.info(f"管理的账户列表: {', '.join([acc.email for acc in accounts])}")
                break
                
            except Exception as e:
                if db:
                    db.rollback()
                    db.close()
                    db = None
                
                error_msg = str(e).lower()
                if 'database is locked' in error_msg or 'locked' in error_msg:
                    if attempt < max_retries - 1:
                        import time
                        delay = base_delay * (2 ** attempt)
                        time.sleep(delay)
                        continue
                
                logger.error(f"启动同步管理器失败: {e}", exc_info=True)
                break
            finally:
                if db:
                    db.close()
        # ========================================
    
    async def stop(self):
        """停止同步管理器"""
        if not self.running:
            return
        
        self.running = False

        # 取消主循环任务
        if self.main_loop_task:
            self.main_loop_task.cancel()
            try:
                await self.main_loop_task
            except asyncio.CancelledError:
                pass
        
        # ========== 停止时状态清理逻辑 ==========
        max_retries = 10
        base_delay = 1.0
        
        managed_account_ids = list(self.managed_accounts)
        
        if managed_account_ids:
            sync_logger.info(f"正在清理 {len(managed_account_ids)} 个账户的同步状态...")
            
            for attempt in range(max_retries):
                db = None
                try:
                    db = SessionLocal()
                    
                    # 批量更新所有处于 syncing 状态的账户
                    updated_count = db.query(EmailAccount).filter(
                        EmailAccount.id.in_(managed_account_ids),
                        EmailAccount.sync_status == "syncing"
                    ).update(
                        {"sync_status": "idle"},
                        synchronize_session=False
                    )
                    db.commit()
                    
                    if updated_count > 0:
                        sync_logger.info(f"已清理 {updated_count} 个账户的同步状态（syncing -> idle）")
                    break
                    
                except Exception as e:
                    if db:
                        db.rollback()
                    
                    error_msg = str(e).lower()
                    if 'database is locked' in error_msg or 'locked' in error_msg:
                        if attempt < max_retries - 1:
                            import time
                            delay = base_delay * (2 ** attempt)
                            time.sleep(delay)
                            continue
                    
                    logger.error(f"清理同步状态失败: {e}", exc_info=True)
                    break
                finally:
                    if db:
                        db.close()
        # ========================================
        
        sync_logger.info(f"同步管理器已停止")
    
    async def add_account(self, account_id: int):
        """添加邮箱账户到同步管理器"""
        async with self._lock:
            self.managed_accounts.add(account_id)
            db = SessionLocal()
            try:
                account = db.query(EmailAccount).filter(EmailAccount.id == account_id).first()
                if account:
                    sync_logger.info(f"【{account.email}】已添加到同步管理器 (ID: {account_id})")
            finally:
                db.close()

    async def remove_account(self, account_id: int):
        """从同步管理器移除邮箱账户"""
        async with self._lock:
            self.managed_accounts.discard(account_id)
            sync_logger.info(f"账户 {account_id} 已从同步管理器移除")
    
    async def _central_sync_loop(self):
        """中心化同步循环，定期检查所有账户"""
        sync_logger.info("中心化同步循环已启动，检查周期: 30秒")
        while self.running:
            try:
                await asyncio.sleep(30)
                
                # 创建一个临时的账户列表副本进行迭代，避免在循环中修改
                accounts_to_check = list(self.managed_accounts)
                if not accounts_to_check:
                    continue

                db = SessionLocal()
                tasks_to_run = []
                try:
                    for account_id in accounts_to_check:
                        account = db.query(EmailAccount).filter(EmailAccount.id == account_id).first()
                        
                        if not account:
                            sync_logger.warning(f"账户 {account_id} 在数据库中未找到，将从管理器中移除")
                            self.managed_accounts.discard(account_id)
                            continue

                        # 条件1: 检查是否被禁用
                        if not account.is_active:
                            continue
                        
                        email_address = account.email
                        sync_type = None
                        
                        # 条件2: 检查是否有 'manual' (手动), 'history' (历史) 消息
                        trigger_key = f"sync_trigger:{account_id}"
                        trigger_data = cache.get(trigger_key)
                        if trigger_data:
                            sync_type = trigger_data.get('type', 'manual')
                            user_id = trigger_data.get('user_id', 'unknown')
                            sync_logger.info(f"【{email_address}】✓ 检测到触发请求 [类型:{sync_type}] [用户ID:{user_id}]")
                            cache.delete(trigger_key) # 清除触发器
                        else:
                            # 条件3: 检查同步时间是否达到
                            fail_count_key = f"imap_fail_count:{account_id}"
                            fail_count = cache.get(fail_count_key) or 0
                            current_sync_interval = 1800 if fail_count >= 3 else self.sync_interval

                            cache_key = f"last_sync_time:{account_id}"
                            last_sync_time = cache.get(cache_key) or 0
                            current_time = get_beijing_time().timestamp()
                            time_since_last_sync = current_time - last_sync_time

                            if time_since_last_sync >= current_sync_interval:
                                sync_type = 'auto'

                        # 如果满足任一同步条件，则创建任务
                        if sync_type:
                            async def sync_wrapper(acc_id, s_type, email):
                                async with self.sync_semaphore:
                                    sync_logger.info(f"【{email}】✓ 获取到同步信号量，开始执行。当前并发数: {5 - self.sync_semaphore._value}/{5}")
                                    success = await self._perform_sync(acc_id, s_type, email)
                                    if success:
                                        # 手动和自动同步都更新时间戳，避免立即触发自动同步
                                        cache_key = f"last_sync_time:{acc_id}"
                                        cache.set(cache_key, get_beijing_time().timestamp())
                                        
                                        # 成功后重置失败计数
                                        fail_count_key = f"imap_fail_count:{acc_id}"
                                        if cache.get(fail_count_key):
                                            cache.delete(fail_count_key)
                                    else:
                                        # 失败后增加失败计数
                                        fail_count_key = f"imap_fail_count:{acc_id}"
                                        fail_count = (cache.get(fail_count_key) or 0) + 1
                                        cache.set(fail_count_key, fail_count, expire=86400) # 24小时
                                        sync_logger.warning(f"【{email}】✗ {s_type}同步失败 (当前连续失败{fail_count}次)")
                                    sync_logger.info(f"【{email}】✓ 同步任务执行完毕，释放信号量。当前并发数: {5 - self.sync_semaphore._value}/{5}")
                            
                            task = asyncio.create_task(sync_wrapper(account_id, sync_type, email_address))
                            tasks_to_run.append(task)
                finally:
                    db.close()
                
                # 并发执行所有需要同步的任务
                if tasks_to_run:
                    sync_logger.info(f"本轮共触发 {len(tasks_to_run)} 个同步任务")
                    await asyncio.gather(*tasks_to_run, return_exceptions=True)

            except asyncio.CancelledError:
                sync_logger.info("中心化同步循环已取消")
                break
            except Exception as e:
                logger.error(f"中心化同步循环出错: {e}", exc_info=True)
                await asyncio.sleep(60) # 发生未知错误时，等待更长时间
        
    async def _perform_sync(self, account_id: int, sync_type: str = "auto", email_address: str = None) -> bool:
        """执行邮件同步，并返回同步是否成功"""
        import ssl
        
        # 获取或创建该账户的同步锁
        if account_id not in self.sync_locks:
            self.sync_locks[account_id] = asyncio.Lock()
        
        sync_lock = self.sync_locks[account_id]
        
        # 检查是否已有同步任务在执行
        if sync_lock.locked():
            if not email_address:
                db = SessionLocal()
                try:
                    account = db.query(EmailAccount).filter(EmailAccount.id == account_id).first()
                    email_address = account.email if account else f"账户{account_id}"
                finally:
                    db.close()
            return False
        
        # 获取同步锁
        async with sync_lock:
            # 检查账户是否仍然启用
            db = SessionLocal()
            try:
                account = db.query(EmailAccount).filter(EmailAccount.id == account_id).first()
                if not account:
                    sync_logger.warning(f"账户 {account_id} 不存在，停止同步")
                    return False
                
                if not account.is_active:
                    sync_logger.info(f"账户 {account_id} 已禁用，停止同步")
                    return False
                
                if not email_address:
                    email_address = account.email
            finally:
                db.close()
            
            # 获取本次同步的开始时间，作为本次同步的标识
            sync_start_time = get_beijing_time().strftime("%H:%M:%S")
        
            # 根据同步类型输出不同的日志
            if sync_type == 'manual':
                sync_logger.info(f"【{email_address}】({sync_start_time}) 开始手动同步")
            elif sync_type == 'history':
                sync_logger.info(f"【{email_address}】({sync_start_time}) 开始历史同步")
            else:
                sync_logger.info(f"【{email_address}】({sync_start_time}) 开始自动同步")
            
            loop = asyncio.get_event_loop()
        
            # 添加重试机制，针对临时网络问题
            max_retries = 2
            retry_delay = 5  # 秒
            
            for attempt in range(max_retries + 1):
                try:
                    new_mail_count, new_mail_objects = await loop.run_in_executor(
                        None,
                        self._sync_account_blocking,
                        account_id,
                        sync_start_time  # 传递同步开始时间
                    )
                    
                    if new_mail_count is not False:
                        sync_end_time = get_beijing_time().strftime("%H:%M:%S")
                        await self._notify_sync_complete(account_id, new_mail_count, sync_type, new_mail_objects)
                        return True
                    else:
                        sync_end_time = get_beijing_time().strftime("%H:%M:%S")
                        sync_logger.warning(f"【{email_address}】({sync_start_time}-{sync_end_time}) 同步失败")
                        await self._notify_sync_failed(account_id)
                        return False
                
                except ssl.SSLEOFError as e:
                    # SSL连接错误，可以重试
                    sync_end_time = get_beijing_time().strftime("%H:%M:%S")
                    if attempt < max_retries:
                        sync_logger.warning(f"【{email_address}】({sync_end_time}) SSL连接错误，{retry_delay}秒后重试 (尝试 {attempt + 1}/{max_retries + 1}): {e}")
                        await asyncio.sleep(retry_delay)
                        continue
                    else:
                        sync_logger.warning(f"【{email_address}】({sync_end_time}) SSL连接错误，已达最大重试次数: {e}")
                        await self._notify_sync_failed(account_id)
                        return False
                
                except Exception as e:
                    sync_end_time = get_beijing_time().strftime("%H:%M:%S")
                    # 检查是否是可重试的错误
                    error_msg = str(e).lower()
                    is_retryable = any(keyword in error_msg for keyword in ['timeout', 'connection', 'network', 'eof'])
                    
                    if is_retryable and attempt < max_retries:
                        sync_logger.warning(f"【{email_address}】({sync_end_time}) 同步出错，{retry_delay}秒后重试 (尝试 {attempt + 1}/{max_retries + 1}): {e}")
                        await asyncio.sleep(retry_delay)
                        continue
                    else:
                        sync_logger.warning(f"【{email_address}】({sync_end_time}) 同步过程出错: {e}")
                        logger.error(f"【{email_address}】({sync_end_time}) 同步过程出错详情", exc_info=True)
                        await self._notify_sync_failed(account_id)
                        return False
            
            return False
    
    def _sync_account_blocking(self, account_id: int, sync_start_time: str = None):
        """阻塞式同步账户（在线程池中运行）"""
        email_service = EmailService(account_id=account_id)
        return email_service.sync_emails(sync_start_time=sync_start_time)
    
    async def _notify_sync_complete(self, account_id: int, new_mail_count: int, sync_type: str, new_mail_objects: list):
        """通知同步完成"""
        import json

        try:
            from api.controller.ws import manager as ws_manager
            
            # 获取账户所属用户
            db = SessionLocal()
            try:
                account = db.query(EmailAccount).filter(EmailAccount.id == account_id).first()
                if account:
                    user_id = account.user_id
                    
                    # 关闭数据库会话
                    db.close()
                    
                    # 统一发送 sync_success 消息
                    message = {
                        "status": "sync_success",
                        "accountId": account_id,
                        "newMailCount": new_mail_count
                    }
                    await ws_manager.send_personal_message(json.dumps(message, default=str), user_id)

                    # 广播未读数更新
                    await ws_manager.broadcast_unread_count_update(user_id)
                else:
                    db.close()
            except Exception:
                db.close()
                raise
        except Exception as e:
            logger.error(f"发送同步完成通知失败: {e}", exc_info=True)
    
    async def _notify_sync_failed(self, account_id: int):
        """通知同步失败"""
        try:
            from api.controller.ws import manager as ws_manager
            
            # 获取账户所属用户
            db = SessionLocal()
            try:
                account = db.query(EmailAccount).filter(EmailAccount.id == account_id).first()
                if account:
                    await ws_manager.send_personal_message(
                        f'{{"status": "sync_failed", "accountId": {account_id}}}',
                        account.user_id
                    )
            finally:
                db.close()
        except Exception as e:
            logger.error(f"发送同步失败通知失败: {e}", exc_info=True)
    
    def get_status(self) -> dict:
        """获取同步管理器状态"""
        return {
            "running": self.running,
            "active_tasks": len(self.managed_accounts),
            "accounts": list(self.managed_accounts),
            "sync_interval": self.sync_interval
        }


# 全局单例实例
sync_manager = EmailSyncManager()
